/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University. All rights reserved.
 * 
 * 
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 * 
 * 1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license; 2) All redistributions in binary form
 * must reproduce the above copyright notice, this list of conditions and the
 * disclaimer listed in this license in the documentation and/or other materials
 * provided with the distribution; 3) Any documentation included with all
 * redistributions must include the following acknowledgement:
 * 
 * "This product includes software developed by the Community Grids Lab. For
 * further information contact the Community Grids Lab at
 * http://communitygrids.iu.edu/."
 * 
 * Alternatively, this acknowledgement may appear in the software itself, and
 * wherever such third-party acknowledgments normally appear.
 * 
 * 4) The name Indiana University or Community Grids Lab or NaradaBrokering,
 * shall not be used to endorse or promote products derived from this software
 * without prior written permission from Indiana University. For written
 * permission, please contact the Advanced Research and Technology Institute
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202. 5) Products
 * derived from this software may not be called NaradaBrokering, nor may Indiana
 * University or Community Grids Lab or NaradaBrokering appear in their name,
 * without prior written permission of ARTI.
 * 
 * 
 * Indiana University provides no reassurances that the source code provided
 * does not infringe the patent or any other intellectual property rights of any
 * other entity. Indiana University disclaims any liability to any recipient for
 * claims brought by any other entity based on infringement of intellectual
 * property rights or otherwise.
 * 
 * LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO
 * WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 * NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF
 * INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS.
 * INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS",
 * "VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.
 * LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR
 * ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION
 * GENERATED USING SOFTWARE.
 */
package cgl.narada.jms;

import java.util.Hashtable;
import java.util.Properties;

import javax.jms.DeliveryMode;
import javax.jms.JMSException;

import cgl.narada.event.NBEvent;
import cgl.narada.event.TemplateInfo;
import cgl.narada.event.TemplateProfileAndSynopsisTypes;
import cgl.narada.matching.Profile;
import cgl.narada.service.ServiceException;
import cgl.narada.service.client.ClientService;
import cgl.narada.service.client.EventConsumer;
import cgl.narada.service.client.EventProducer;
import cgl.narada.service.client.NBEventListener;
import cgl.narada.service.client.NBRecoveryListener;
import cgl.narada.service.client.NBRecoveryNotification;
import cgl.narada.service.client.SessionService;
import cgl.narada.service.qos.ConsumerConstraints;
import cgl.narada.service.qos.ProducerConstraints;
import cgl.narada.util.Counter;

/**
 * This class provides the bridge for JMS calls into the NaradaBrokering system.
 * 
 * @author Shrideep Pallickara $Revision$ $Date$
 */

public class NBJmsInterconnectionBridge implements NBEventListener,
    NBRecoveryListener {

    private ClientService clientService;

    private EventProducer producer;

    /**
     * Maintain a reference to the JmsTopicConnection that the data packets
     * would be routed to. This topicConnection then deal with the whole process
     * of start(), stop(), pause() or what have you that JMS requires support
     * for. Individual topicConnections then route packets to individual
     * sessions
     */
    private JmsTopicConnection jmsTopicConnection;

    private Properties bridgeProperties;

    private Hashtable producerConstraints;

    private Hashtable recoveringTemplates;

    private Hashtable recoveryStatus;

    private Hashtable profiles;

    private Hashtable subscriptionCount;

    private Hashtable consumers;

    private String moduleName = "NBJmsInterconnectionBridge: ";

    public NBJmsInterconnectionBridge(Properties bridgeProperties,
        String commType, int entityId) throws JMSException {
        try {
            this.bridgeProperties = bridgeProperties;
            clientService = SessionService.getClientService(entityId);
            
            if (bridgeProperties.containsKey("IPSecTunnel")) {
                clientService.initializeBrokerCommunications(bridgeProperties,
                                                             "ipsec");
            }
            
            clientService.initializeBrokerCommunications(bridgeProperties,
                                                         commType);

            producerConstraints = new Hashtable();
            recoveringTemplates = new Hashtable();
            recoveryStatus = new Hashtable();

            profiles = new Hashtable();
            subscriptionCount = new Hashtable();
            consumers = new Hashtable();

        } catch (ServiceException serEx) {
            throw new JMSException(moduleName + serEx.toString());
        }
    }

    public void setJmsTopicConnection(JmsTopicConnection jmsTopicConnection) {
        this.jmsTopicConnection = jmsTopicConnection;
    }

    public synchronized void unSubscribeConsumer(String subscription)
        throws JMSException {
        int count = decrementSubscriptionCount(subscription);

        if (count != 0) {
            System.out.println(moduleName + "Subscription Count = " + count
                + ". Will not propagate unsubscribe() since there are other"
                + " active subscriptions to this topic.");
            return;
        }

        try {
            Profile profile = clientService
                .createProfile(TemplateProfileAndSynopsisTypes.STRING,
                               subscription);

            String profileId = profile.getProfileId();
            if (!consumers.containsKey(profileId)) {
                String errorReport = moduleName
                    + "the consumer associated with " + " the subscription ["
                    + subscription + "] cannot be located.";
                System.out.println(errorReport);
                return;
            }

            EventConsumer consumer = (EventConsumer) consumers
                .remove(profileId);
            consumer.unSubscribe(profile);
            profiles.remove(profileId);
            
        } catch (Exception serEx) {
            throw new JMSException(moduleName + "Problems unsubscribing"
                + serEx.toString());
        }
    }

    public synchronized void initializeConsumer(String subscription,
                                                boolean isDurable)
        throws JMSException {

        try {
            Profile profile = clientService
                .createProfile(TemplateProfileAndSynopsisTypes.STRING,
                               subscription);            
            String profileId = profile.getProfileId();
            if (profiles.containsKey(profileId)) {
                incrementSubscriptionCount(subscription);
                return;
            }

            

            EventConsumer consumer = clientService.createEventConsumer(this);
            consumers.put(profileId, consumer);

            if (!isDurable) {
                consumer.subscribeTo(profile);
                incrementSubscriptionCount(subscription);
            } else {
                int templateId = getTemplateId(subscription, true);
                ConsumerConstraints constraints = consumer
                    .createConsumerConstraints(profile);
                constraints.setReceiveReliably(templateId);
                consumer.subscribeTo(profile, constraints);
                incrementSubscriptionCount(subscription);
                long recoveryId = consumer.recover(templateId, this);
                waitAndProcessRecoveryResponse(recoveryId, "subscriber");
            }

            profiles.put(profileId, profile);
        } catch (Exception serEx) {
            throw new JMSException(moduleName + serEx.toString());
        }
    }

    private int incrementSubscriptionCount(String subscription) {
        if (!subscriptionCount.containsKey(subscription)) {
            subscriptionCount.put(subscription, new Counter());
        }

        Counter count = (Counter) subscriptionCount.get(subscription);
        return count.incrementCount();
    }

    private int decrementSubscriptionCount(String subscription) {
        if (!subscriptionCount.containsKey(subscription)) {
            String errorReport = moduleName + "Is not aware of subscription "
                + "[" + subscription
                + "]. No further actions will be performed.";
            System.out.println(errorReport);
        }

        Counter count = (Counter) subscriptionCount.get(subscription);
        return count.decrementCount();
    }

    public void publishEvent(JmsEvent jmsEvent) throws JMSException {
        try {
            String topic = jmsEvent.getTopic();
            int deliveryMode = jmsEvent.getDeliveryMode();
            int templateId;
            ProducerConstraints constraints = null;
            if (producer == null) {
                producer = clientService.createEventProducer();
            }

            if (deliveryMode == DeliveryMode.NON_PERSISTENT) {
                // templateId = getTemplateId(topic, false);
                templateId = topic.hashCode();
            } else {
                templateId = getTemplateId(topic, true);
                constraints = getProducerConstraints(templateId, topic);
            }/* end if */

            producer.generateEventIdentifier(true);
            producer.setTemplateId(templateId);
            producer.setDisableTimestamp(false);

            NBEvent nbEvent = producer
                .generateEvent(TemplateProfileAndSynopsisTypes.STRING, topic,
                               jmsEvent.getBytes());

            if (deliveryMode == DeliveryMode.NON_PERSISTENT) {
                producer.publishEvent(nbEvent);
            } else {
                producer.publishEvent(nbEvent, constraints);
            }

        } catch (ServiceException serEx) {
            throw new JMSException(moduleName + serEx.toString());
        }
    }

    public void onEvent(NBEvent nbEvent) {
        // System.out.println(moduleName + "Received NBEvent ");

        byte[] data = nbEvent.getContentPayload();
        if (jmsTopicConnection != null) {
            jmsTopicConnection.processDataReceived(data);
        } else {
            System.out.println("The JmsTopicConnection is Null");
        }
    }

    /**
     * Upon completion of the attempt to recover, this method is invoked on the
     * listener that was registered with the
     */
    public void onRecovery(NBRecoveryNotification recoveryNotification) {
        long recoveryId = recoveryNotification.getRecoveryId();
        Long recoveryKey = new Long(recoveryId);

        if (recoveryStatus.containsKey(recoveryNotification)) {
            System.out.println(moduleName
                + "Duplicate recovery Notificiation received!");
            return;
        }

        recoveryStatus.put(recoveryKey, recoveryNotification);
        /**
         * Proceed to wakeup piece where we are waiting for this notification to
         * proceed
         */
        Object syncObject = recoveringTemplates.get(recoveryKey);
        synchronized (syncObject) {
            syncObject.notify();
        }
    }

    private void waitAndProcessRecoveryResponse(long recoveryId, String pubOrSub)
        throws JMSException {
        Long recoveryKey = new Long(recoveryId);

        /** Proceed to wait for a response */
        Object syncObject = new Object();
        recoveringTemplates.put(recoveryKey, syncObject);
        synchronized (syncObject) {
            System.out.println(moduleName + "Waiting for " + pubOrSub
                + " recovery");
            try {
                syncObject.wait();
            } catch (InterruptedException ie) {
                throw new JMSException(
                                       "Error while waiting for recovery response "
                                           + ie.toString());
            }
        }
        ; /* end synchronized */

        System.out.println(moduleName + "Waking up to process response ");
        NBRecoveryNotification recoveryNotification = (NBRecoveryNotification) recoveryStatus
            .get(recoveryKey);
        boolean success = recoveryNotification.isSuccessful();

        if (!success) {
            throw new JMSException(moduleName + recoveryNotification.toString());
        } else {
            System.out.println(moduleName + recoveryNotification + "\n"
                + pubOrSub + " recovery complete !");
        }/* end if (success) */

    }

    private ProducerConstraints getProducerConstraints(int templateId,
                                                       String topic)
        throws JMSException {

        try {
            Object templateKey = new Integer(templateId);
            ProducerConstraints constraints;
            if (!producerConstraints.containsKey(templateKey)) {
                TemplateInfo templateInfo = clientService
                    .createTemplateInfo(templateId,
                                        TemplateProfileAndSynopsisTypes.STRING,
                                        topic);
                constraints = producer.createProducerConstraints(templateInfo);

                /** Proceed to recover on the template */
                long recoveryId = producer.recover(templateId, this);
                waitAndProcessRecoveryResponse(recoveryId, "publisher");

                producerConstraints.put(templateKey, constraints);
            }
            return (ProducerConstraints) producerConstraints.get(templateKey);
        } catch (Exception serEx) {
            throw new JMSException(serEx.toString());
        }
    }

    /** This method retrievs the templateId associated with a given topic */
    private int getTemplateId(String topic, boolean reliable)
        throws JMSException {
        int templateId = 0;
        boolean notemplateId = false;
        String reason = "";

        String templateString = bridgeProperties.getProperty(topic);
        if (templateString == null) {
            notemplateId = true;
            reason = "No templateID specified for the reliable topic [" + topic
                + "]. ";
        } else {
            try {
                templateId = Integer.parseInt(templateString);
            } catch (NumberFormatException nfe) {
                notemplateId = true;
                reason = "The templateID specified for the reliable topic ["
                    + topic + "] is (" + templateString + "), which gives a "
                    + "NumberFormatException. Please check the value specified";
            } /* end try-catch */
        } /* end if */

        if (notemplateId && reliable) {
            throw new JMSException(reason + "NaradaBrokering needs this for "
                + "reliable interactions");
        }

        /** We can get around the problem of templateId for unreliable topics */
        if (!reliable && templateId == 0) {
            templateId = topic.hashCode();
        }

        if (reliable && templateId == 0) {
            throw new JMSException(
                                   "TemplateId specified is (0). Please specify a valid"
                                       + " templateId. NaradaBrokering needs this for "
                                       + "reliable interactions");
        }

        return templateId;
    }

    public void closeConnection() throws JMSException {
        try {
            clientService.closeBrokerConnection();

            clientService.terminateServices();

            try {
                /**
                 * Make sure that the transmission tracker is shutdown
                 * gracefully.
                 */
                Thread.sleep(100);
            } catch (InterruptedException e) {
                System.out
                    .println(moduleName
                        + "Problems sleeping while waiting for transfer completion");
            }

        } catch (ServiceException serEx) {
            throw new JMSException(serEx.toString());
        }
    }

}
